package com.beiding.jsonanalyzer.datasource;

import com.alibaba.fastjson.JSONObject;
import com.beiding.jsonanalyzer.flow.*;
import com.beiding.jsonanalyzer.flow.branchevents.FlowEnd;
import com.beiding.jsonanalyzer.flow.branchevents.StopNow;
import com.beiding.jsonanalyzer.flow.flowevents.BatchEndEvent;
import com.beiding.jsonanalyzer.flow.flowevents.BatchStartEvent;
import com.beiding.jsonanalyzer.flow.flowevents.DataNodeEvent;
import com.csvreader.CsvReader;

import java.io.Reader;

/**
 * @author 丁常磊
 * @date 2022/5/16 10:01
 */

public class CsvDatasource extends DataSource<CsvDatasource.CsvRow> {


    private Reader reader;

    private char cellSplit;


    public CsvDatasource(Reader reader, char cellSplit) {
        super("csv", CsvRow::new);
        this.reader = reader;
        this.cellSplit = cellSplit;
    }

    public CsvDatasource(Reader reader) {
        this(reader, ',');
    }

    @Override
    protected void flowStart(BranchNode<CsvRow> branchNode, Flow flow) {


        boolean[] stop = new boolean[]{false};
        //记录异常
        StopNow[] stopNows = new StopNow[]{null};
        branchNode.getBranch().addListener(StopNow.class, e -> {
            stopNows[0] = e;
            stop[0] = true;
        });

        try {
            CsvReader csvReader = new CsvReader(reader, cellSplit);

            Object batch = DefaultBatch.INSTANCE;

            branchNode.signDataEvent(new BatchStartEvent<>(batch));
            if (csvReader.readHeaders()) {
                long row = 0;
                while (csvReader.readRecord()) {

                    if (stop[0]) {
                        break;
                    }


                    JSONObject jsonObject = new JSONObject();
                    for (int i = 0; i < csvReader.getHeaderCount(); i++) {
                        String header = csvReader.getHeader(i);
                        String value = csvReader.get(i);
                        jsonObject.put(header, value);
                    }

                    CsvRow valueNode = new CsvRow();
                    valueNode.rowNum = row++;
                    valueNode.set(jsonObject);

                    branchNode.signDataEvent(new DataNodeEvent<>(valueNode));

                }
            }

            branchNode.signDataEvent(new BatchEndEvent<>(batch));

            if (stopNows[0] != null) {
                flow.emit(new FlowEnd(stopNows[0].getMsg(), stopNows[0].getException()));
            } else {
                branchNode.flush(() -> {
                    flow.emit(new FlowEnd("执行成功", null));
                });
            }

        } catch (Exception e) {
            branchNode.flush(() -> {
                flow.emit(new FlowEnd("执行失败", e));
            });
        }

    }

    @Override
    public void close() {
        try {
            this.reader.close();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }


    public static class CsvRow extends ValueNode{
        private long rowNum;

        public long getRowNum() {
            return rowNum;
        }
    }

}
